package org.example.hbase2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * namespace
 * exists
 * create table
 * drop table
 */
public class HBaseDDL {

    private final Logger logger = LoggerFactory.getLogger(HBaseDDL.class);

    private Configuration conf;
    private Connection con;
    private Admin admin;

//    static {
//
//    }

    @Before
    public void init() throws IOException {
        // conf
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop1, hadoop2, hadoop3");
        // connector
        con = ConnectionFactory.createConnection(conf);
        // ddl object
        admin = con.getAdmin();
        // operation
    }

    @After
    public void uninit() {
        try {
            admin.close();
            con.close();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
    }

    @Test
    public void testCreate() throws IOException {
        createNamespace("big");
    }

    // TODO create namespace
    public void createNamespace(String ns) throws IOException {
        NamespaceDescriptor nsd = NamespaceDescriptor.create(ns).build();
        admin.createNamespace(nsd);
    }


    public void isTableExists(String tn) throws IOException {
        admin.tableExists(TableName.valueOf(tn));
    }


    public void createTable(String tn, String... cf) throws IOException {
        Assert.assertNotNull(tn);
        if (cf.length <= 0) {
            logger.info("zero column family.");
            return;
        }
        TableDescriptorBuilder tdb = TableDescriptorBuilder.newBuilder(TableName.valueOf(tn));
        List<ColumnFamilyDescriptor> cfds = new ArrayList<>();
        for (String s : cf) {
            ColumnFamilyDescriptor cfd =
                    ColumnFamilyDescriptorBuilder
                            .newBuilder(Bytes.toBytes(s))
//                    .setMaxVersions(3)
//                    .setBlocksize(128)
                    .build();
            cfds.add(cfd);
        }
        tdb.setColumnFamilies(cfds);
        tdb.setCoprocessor("org.example.hbase2.HBase_Coprocessor"); // 注册协处理器
        TableDescriptor td = tdb.build();
        admin.createTable(td);
    }


    public void deleteTable(String tn) throws IOException {
        if (admin.tableExists(TableName.valueOf(tn))) {
            admin.disableTable(TableName.valueOf(tn));
            admin.deleteTable(TableName.valueOf(tn));
        }
    }


    public void putData(String tn, String rowKey, String cf, String cn, String value) throws IOException {
        Table table = con.getTable(TableName.valueOf(tn));
        Put put = new Put((Bytes.toBytes(rowKey)));
        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cf),Bytes.toBytes(cf));
        table.put(put);
    }
}
